package mq

import (
	"context"
	"encoding/json"
	"errors"

	"github.com/apache/rocketmq-client-go/v2/rlog"

	"gitlab.vspncd.com/mic/common/mq/internal/rocket"
)

type DelayLevel int

const (
	// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
	DelayLevel1s  DelayLevel = 1
	DelayLevel5s  DelayLevel = 2
	DelayLevel10s DelayLevel = 3
	DelayLevel30s DelayLevel = 4
	DelayLevel1m  DelayLevel = 5
	DelayLevel2m  DelayLevel = 6
	DelayLevel3m  DelayLevel = 7
	DelayLevel4m  DelayLevel = 8
	DelayLevel5m  DelayLevel = 9
	DelayLevel6m  DelayLevel = 10
	DelayLevel7m  DelayLevel = 11
	DelayLevel8m  DelayLevel = 12
	DelayLevel9m  DelayLevel = 13
	DelayLevel10m DelayLevel = 14
	DelayLevel20m DelayLevel = 15
	DelayLevel30m DelayLevel = 16
	DelayLevel1h  DelayLevel = 17
	DelayLevel2h  DelayLevel = 18
)

type AuthConfig = rocket.AuthConfig

type (
	RocketMq interface {
		// 开启
		Start()
		// 关闭
		Stop()
		// 设置链路
		SetUp(config Config)
		// 注册消费者
		RegConsumer(topic string, handle func(key string, body []byte))
		// 发送延迟消息
		SendDelayMsg(topic string, key string, v interface{}, delayLevel DelayLevel) (string, error)
		// 发送消息
		SendMsg(topic string, v interface{}) (string, error)
		// 发送消息，带key
		SendMsgWithKey(topic string, key string, v interface{}) (string, error)
	}

	defaultRocketMq struct {
		mode     string // 启动模式
		consumes map[string]rocket.Consumer
		pushes   map[string]rocket.Producer
	}

	Config struct {
		Addr  []string    `json:"addr"` // 127.0.0.1:9876
		Topic string      `json:"topic"`
		Auth  *AuthConfig `json:"auth"`
	}
)

func NewRocketMq(mode ...string) RocketMq {
	target := &defaultRocketMq{
		consumes: map[string]rocket.Consumer{},
		pushes:   map[string]rocket.Producer{},
	}

	if len(mode) > 0 {
		target.mode = mode[0]
	}

	rlog.SetLogger(rocket.NewLogger())

	return target
}

func (d *defaultRocketMq) SetUp(config Config) {
	if _, ok := d.pushes[config.Topic]; ok {
		panic("已存在相同的topic监听")
	}
	d.pushes[config.Topic] = rocket.NewProducer(&rocket.ProducerConfig{
		Addr:  config.Addr,
		Topic: config.Topic,
		Auth:  config.Auth,
	})
	d.consumes[config.Topic] = rocket.NewConsumer(&rocket.ConsumerConfig{
		Addr:  config.Addr,
		Topic: config.Topic,
		Auth:  config.Auth,
	})
}

func (d *defaultRocketMq) RegConsumer(topic string, handle func(key string, body []byte)) {
	if _, ok := d.consumes[topic]; !ok {
		panic("不存在topic监听")
	}
	d.consumes[topic].Consume(handle)
}

func (d *defaultRocketMq) Start() {
	for _, c := range d.consumes {
		c.Start()
	}
	for _, p := range d.pushes {
		p.Start()
	}
}

func (d *defaultRocketMq) Stop() {
	for _, p := range d.pushes {
		p.Stop()
	}
	for _, c := range d.consumes {
		c.Stop()
	}
}

// 发送延迟消息
func (d *defaultRocketMq) SendDelayMsg(topic string, key string, v interface{}, delayLevel DelayLevel) (string, error) {
	p, body, err := d.sendCheck(topic, v)
	if err != nil {
		return "", err
	}
	return p.SendDelayMsg(context.Background(), key, body, int(delayLevel))
}

func (d *defaultRocketMq) SendMsg(topic string, v interface{}) (string, error) {
	p, body, err := d.sendCheck(topic, v)
	if err != nil {
		return "", err
	}
	return p.SendMsg(context.Background(), body)
}

func (d *defaultRocketMq) SendMsgWithKey(topic string, key string, v interface{}) (string, error) {
	p, body, err := d.sendCheck(topic, v)
	if err != nil {
		return "", err
	}
	return p.SendMsgWithKey(context.Background(), key, body)
}

func (d *defaultRocketMq) sendCheck(topic string, v interface{}) (rocket.Producer, []byte, error) {
	p, ok := d.pushes[topic]
	if !ok {
		return nil, nil, errors.New("topic is not exist")
	}

	body, err := json.Marshal(v)
	if err != nil {
		return nil, nil, err
	}

	return p, body, nil
}
